package com.wpf.springbootflux.flux;


import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

/**
 * reactor = jdk8 stream + jdk9 reactive stream
 * Mono 0-1 个元素
 * Flux 0-n 个元素
 */
@Slf4j
@RestController
public class FluxDemo {


    @GetMapping("get1")
    private String get1(){
        log.info("普通的web访问！");
        return "get1";
    }


    /**
     * Mono是返回的是 [0-1]
     * @return
     */
    @GetMapping("get2")
    private Mono<String> get2(){
        log.info("Mono 的web访问！");
        return Mono.just("get2");
    }


//    @GetMapping(value = "get3", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    @GetMapping(value = "get3", produces = "text/event-stream")
    private Flux<String> get3(){
        log.info("Flux 产生数据！");

        Flux<String> result = Flux.fromStream(IntStream.range(0, 5).mapToObj(s -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "flux data -->" + s;
        }));

        return result;
    }


//    public static void main(String[] args) {
//
//        Subscriber<Integer> subscriber = new Subscriber<Integer>() {
//
//            private Subscription subscription;
//
//            @Override
//            public void onSubscribe(Subscription subscription) {
//                // 保存订阅关系, 需要用它来给发布者响应
//                this.subscription = subscription;
//
//                // 请求一个数据
//                this.subscription.request(1);
//            }
//
//            @Override
//            public void onNext(Integer item) {
//                // 接受到一个数据, 处理
//                System.out.println("接受到数据: " + item);
//
//                try {
//                    TimeUnit.SECONDS.sleep(3);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
//
//                // 处理完调用request再请求一个数据
//                this.subscription.request(1);
//
//                // 或者 已经达到了目标, 调用cancel告诉发布者不再接受数据了
//                // this.subscription.cancel();
//            }
//
//            @Override
//            public void onError(Throwable throwable) {
//                // 出现了异常(例如处理数据的时候产生了异常)
//                throwable.printStackTrace();
//
//                // 我们可以告诉发布者, 后面不接受数据了
//                this.subscription.cancel();
//            }
//
//            @Override
//            public void onComplete() {
//                // 全部数据处理完了(发布者关闭了)
//                System.out.println("处理完了!");
//            }
//
//        };
//
//        String[] arr = {"111","222","333","444","555"};
//        // jdk8 stream
//        Flux.fromArray(arr).map(x->Integer.parseInt(x))
//
//                // jdk9 reactive stream
//                .subscribe(subscriber);
//
//    }

}
